package webapi

import (
	"MqttCloudProtocolServer/publiclib/dbcontrol"
	"MqttCloudProtocolServer/publiclib/helplib"
	"fmt"
	"net/http"
	"strconv"
	"sync"
	"time"

	mqtt "github.com/eclipse/paho.mqtt.golang"
	"github.com/wonderivan/logger"
)

type UserThread struct {
	key         string
	uuid        string
	lockFrame   []string
	s_topic     string
	r_topic     string
	old_s_topic string
	old_r_topic string
	addr        string
	port        int
	user        string
	pwd         string
	framelocker sync.Mutex
	timelocker  sync.Mutex
	time        time.Time
	limits      int
	mqttClient  mqtt.Client
	textMode    bool
	isEnd       bool
}

func (_this *UserThread) Start(limitsParam int) {
	_this.isEnd = false
	_this.limits = limitsParam
	addr := "tcp://" + _this.addr + ":" + strconv.Itoa(_this.port)
	opts := mqtt.NewClientOptions().AddBroker(addr).SetClientID(_this.key + "_clientWatch")

	//set userName
	opts.SetUsername(_this.user)
	opts.SetPassword(_this.pwd)
	opts.SetKeepAlive(2 * time.Second)
	opts.SetPingTimeout(1 * time.Second)
	opts.SetConnectRetryInterval(time.Second * 1)
	opts.SetConnectRetry(true)
	opts.SetAutoReconnect(true)
	opts.SetMaxReconnectInterval(time.Second * 1)
	opts.ResumeSubs = true
	opts.SetOnConnectHandler(func(cc mqtt.Client) {
		logger.Debug(_this.key + ": Connect")
		_this.mqttClient = cc
		//subscribe topic
		token := _this.mqttClient.Subscribe(_this.r_topic, 0, _this.ReceiveFrame)
		if token.Error() != nil {
			logger.Alert("token.Error()")
			logger.Alert(token.Error().Error())
			token.Wait()
		}
		token = _this.mqttClient.Subscribe(_this.s_topic, 0, _this.ReceiveFrame)
		if token.Error() != nil {
			logger.Alert("token.Error()")
			logger.Alert(token.Error().Error())
			token.Wait()
		}
	})
	opts.SetConnectionLostHandler(func(_ mqtt.Client, _ error) {
		_this.mqttClient = nil
		logger.Debug(_this.key + ": LostConnect")
	})

	opts.SetReconnectingHandler(func(_ mqtt.Client, _ *mqtt.ClientOptions) {
		_this.ReSubs()
		logger.Debug(_this.key + ": ReConnect")
	})
	_this.mqttClient = mqtt.NewClient(opts)
	if token := _this.mqttClient.Connect(); token.Wait() && token.Error() != nil {
		logger.Alert("token.Error()")
		logger.Alert(token.Error().Error())
	}

	go func() {
		for {
			now := time.Now()
			_this.timelocker.Lock()
			last := _this.time
			_this.timelocker.Unlock()
			diff := now.Sub(last).Seconds()
			if diff > float64(_this.limits) {
				_this.mqttClient.Disconnect(500)
				_this.mqttClient = nil
				_this.isEnd = true
				return
			}
			time.Sleep(1 * time.Second)
		}
	}()
}
func (_this *UserThread) Change() {
	_this.ReSubs()
}

func (_this *UserThread) ReSubs() {
	if _this.old_r_topic != _this.r_topic {
		if len(_this.old_r_topic) > 0 {
			_this.mqttClient.Unsubscribe(_this.old_r_topic)
		}
		token := _this.mqttClient.Subscribe(_this.r_topic, 0, _this.ReceiveFrame)
		if token.Error() != nil {
			logger.Alert("token.Error()")
			logger.Alert(token.Error().Error())
			token.Wait()
		}
		_this.old_r_topic = _this.r_topic
	}
	if _this.old_s_topic != _this.s_topic {
		if len(_this.old_s_topic) > 0 {
			_this.mqttClient.Unsubscribe(_this.old_s_topic)
		}
		token := _this.mqttClient.Subscribe(_this.s_topic, 0, _this.ReceiveFrame)
		if token.Error() != nil {
			logger.Alert("token.Error()")
			logger.Alert(token.Error().Error())
			token.Wait()
		}
		_this.old_s_topic = _this.s_topic
	}
}
func (_this *UserThread) Keepalive() {
	_this.timelocker.Lock()
	_this.time = time.Now()
	_this.timelocker.Unlock()
}
func (_this *UserThread) ReceiveFrame(client mqtt.Client, message mqtt.Message) {
	topic := message.Topic()
	payload := message.Payload()
	frame := ""
	if _this.textMode {
		frame = "{" + "\"topic\":\"" + topic + "\",\"value\":\"" + string(payload) + "\"}"
	} else {
		frame = "{" + "\"topic\":\"" + topic + "\",\"value\":\"" + helplib.BufferToString(payload) + "\"}"
	}
	_this.framelocker.Lock()
	if len(_this.lockFrame) == cap(_this.lockFrame) {
		_this.lockFrame = _this.lockFrame[1:]
	}
	_this.lockFrame = append(_this.lockFrame, frame)
	_this.framelocker.Unlock()
}

//GatewayWatcherService 网关服务器注册API类。注册控制台
type GatewayWatcherService struct {
	base       ServiceBase
	threadPoll map[string]*UserThread
}

//Regist 注册
func (_this *GatewayWatcherService) Regist(mux *http.ServeMux) {
	_this.base.Regist(mux)
	_this.base.path += "gsw"
	_this.base.HandleRegist("getServerIP", _this.getServerIP)
	_this.base.HandleRegist("getFrame", _this.getFrame)
	_this.threadPoll = make(map[string]*UserThread)
}

//getServerIP 增加一个网关连接请求
func (_this *GatewayWatcherService) getServerIP(w http.ResponseWriter, r *http.Request) {
	vars := r.URL.Query()
	if len(vars) > 0 {
		uuid := GetParam(vars, "uuid")
		control := _this.GetControl(uuid)
		if control == nil {
			fmt.Fprintln(w, "Failed")
			return
		}
		fmt.Fprintln(w, control.ParentTCPConf.ParentMqttServer.Data.Host)
	}
}

//getFrame 获取缓存的帧
func (_this *GatewayWatcherService) getFrame(w http.ResponseWriter, r *http.Request) {
	vars := r.URL.Query()
	if len(vars) > 0 {
		user := GetParam(vars, "user")
		uuid := GetParam(vars, "uuid")
		if _, ok := _this.threadPoll[user]; ok {
			if _this.threadPoll[user].isEnd {
				delete(_this.threadPoll, user)
			}
		}
		if _, ok := _this.threadPoll[user]; !ok {
			control := _this.GetControl(uuid)
			if control == nil {
				fmt.Fprintln(w, "Failed")
				return
			}
			//创建线程
			_this.threadPoll[user] = &UserThread{
				uuid:        uuid,
				key:         user,
				lockFrame:   make([]string, 0, 20),
				s_topic:     control.Data.PubTopic,
				r_topic:     control.Data.SubTopic,
				old_s_topic: control.Data.PubTopic,
				old_r_topic: control.Data.PubTopic,
				user:        control.ParentTCPConf.ParentMqttServer.Data.User,
				pwd:         control.ParentTCPConf.ParentMqttServer.Data.Password,
				addr:        control.ParentTCPConf.ParentMqttServer.Data.Host,
				port:        control.ParentTCPConf.ParentMqttServer.Data.Port,
				isEnd:       false,
				time:        time.Now(),
				textMode:    control.ParentTCPConf.Data.Protocol > 5 && control.ParentTCPConf.Data.Protocol != 7,
			}
			_this.threadPoll[user].Start(10)
			_this.threadPoll[user].Keepalive()
		} else {
			_this.threadPoll[user].Keepalive()
			if _this.threadPoll[user].uuid != uuid {
				_this.threadPoll[user].uuid = uuid
				_this.threadPoll[user].framelocker.Lock()
				_this.threadPoll[user].lockFrame = _this.threadPoll[user].lockFrame[0:0]
				control := _this.GetControl(uuid)
				if control == nil {
					fmt.Fprintln(w, "Failed")
					_this.threadPoll[user].framelocker.Unlock()
					return
				}
				_this.threadPoll[user].s_topic = control.Data.PubTopic
				_this.threadPoll[user].r_topic = control.Data.SubTopic
				_this.threadPoll[user].Change()
				_this.threadPoll[user].textMode = control.ParentTCPConf.Data.Protocol > 5 && control.ParentTCPConf.Data.Protocol != 7
				_this.threadPoll[user].framelocker.Unlock()
				//UUID改变
			} else {
				if _this.threadPoll[user].lockFrame != nil && len(_this.threadPoll[user].lockFrame) > 0 {
					_this.threadPoll[user].framelocker.Lock()
					willSend := "["
					for index, frame := range _this.threadPoll[user].lockFrame {
						if index > 0 {
							willSend += ","
						}
						willSend += frame
					}
					willSend += "]"
					fmt.Fprintln(w, willSend)
					_this.threadPoll[user].lockFrame = _this.threadPoll[user].lockFrame[0:0]
					_this.threadPoll[user].framelocker.Unlock()
				}
			}
		}
	}
}

func (*GatewayWatcherService) GetControl(uuid string) *dbcontrol.MqttconfControl {
	control := dbcontrol.CreateMqttconfControl()
	if control == nil {
		logger.Alert("无法获取数据库控制器")
		return nil
	}
	if err := control.FindByID(uuid); err != nil {
		logger.Alert(err)
		return nil
	}
	if err := control.GetParentTCPConf(); err != nil {
		logger.Alert(err)
		return nil
	}
	if err := control.ParentTCPConf.GetParentMqttServer(); err != nil {
		logger.Alert(err)
		return nil
	}
	return control
}
